package org.jetlinks.pro.things.impl.connector;

import org.jetlinks.core.message.ThingMessage;
import org.jetlinks.core.things.MetadataId;
import org.jetlinks.core.things.ThingId;
import org.jetlinks.pro.things.ThingsConnector;
import reactor.core.Disposable;
import reactor.core.Disposables;
import reactor.core.publisher.Mono;

import java.util.function.Function;

/**
 * 抽象物连接器，默认实现dispose相关功能
 *
 * @author zhouhao
 * @since 1.12
 */
public abstract class AbstractThingsConnector implements ThingsConnector {

    //申明一个可以释放响应式流资源的容器。
    protected final Disposable.Composite disposeActions = Disposables.composite();

    /**
     * 当dispose时此方法被执行,可用于进行资源释放等操作
     */
    protected abstract void onDispose();

    /**
     * 连接到当前物的物模型.物实例ID由数据源生成. 通常用于批量连接处理,比如在物模版里配置规则,计算对应物实例
     * 当产生消息时,handler将被调用.可调用返回值的{@link Disposable#dispose()}来断开连接
     *
     * @param metadataId 物模型ID
     * @param handler    消息处理器
     * @return Disposable
     */
    @Override
    public final Disposable connectTo(MetadataId metadataId, Function<ThingMessage, Mono<Void>> handler) {
        Disposable disposable = doConnectTo(metadataId, handler);
        if(!disposable.isDisposed()) {
            doOnDispose(disposable);
        }
        return disposable;
    }

    /**
     * 连接到指定物的物模型,当产生消息时,handler将被调用.可调用返回值的{@link Disposable#dispose()}来断开连接
     *
     * @param thingId    物ID
     * @param metadataId 物模型ID
     * @param handler    消息处理器
     * @return Disposable
     */
    @Override
    public final Disposable connectTo(ThingId thingId, MetadataId metadataId, Function<ThingMessage, Mono<Void>> handler) {
        Disposable disposable = doConnectTo(thingId,metadataId, handler);
        if(!disposable.isDisposed()) {
            doOnDispose(disposable);
        }
        return disposable;
    }

    protected abstract Disposable doConnectTo(MetadataId metadataId, Function<ThingMessage, Mono<Void>> handler);

    protected abstract Disposable doConnectTo(ThingId thingId, MetadataId metadataId, Function<ThingMessage, Mono<Void>> handler);

    /**
     * 监听断开连接,并执行自定义动作
     *
     * @param action 自定义动作
     */
    @Override
    public final void doOnDispose(Disposable action) {
        if (isDisposed()) {
            action.dispose();
            return;
        }
        disposeActions.add(action);
    }

    /**
     * 当资源或任务被释放时，返回true。
     * @return 资源是否被释放的bool量
     */
    @Override
    public final boolean isDisposed() {
        return disposeActions.isDisposed();
    }

    /**
     * 释放任务资源
     */
    @Override
    public final void dispose() {
        disposeActions.dispose();
        onDispose();
    }

}
